﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Reflection;
using PServiceBus.Core.Runtime.Extensions;
using Phoenix.ReflectionExtensions;
using PServiceBus.Core.Runtime;
using System.Threading;
using PServiceBus.Services.Gateway.Runtime;
using System.Configuration;
using PServiceBus.Services.Gateway.Extensions;
using PServiceBus.Core.Logger;
using System.Collections.Concurrent;
using PServiceBus.Core.Interface;
using PServiceBus.Gateway.Interfaces;
using PServiceBus.Core.Manager;

namespace PServiceBus.WebSocket {
    public class WebSocketHandler {
        internal static ConcurrentDictionary<string, IMessageBus> _messageBuses = new ConcurrentDictionary<string, IMessageBus>();
        internal static ConcurrentDictionary<string, ConcurrentDictionary<string, dynamic>> _handlers = new ConcurrentDictionary<string, ConcurrentDictionary<string, dynamic>>();
        internal static ConcurrentDictionary<Guid, string> _connections = new ConcurrentDictionary<Guid, string>();
        internal static ConcurrentDictionary<string, string> _connectionTokens = new ConcurrentDictionary<string, string>();
        internal static ConcurrentDictionary<string, Dictionary<string, string>> _connectionHeaders = new ConcurrentDictionary<string, Dictionary<string, string>>();
        internal static ConcurrentDictionary<string, IWebSocketConnection> _webSockets = new ConcurrentDictionary<string, IWebSocketConnection>();
        internal static string _esbEndpointAddress;
        const int WEBSOCKET_RETRY_MAX = 5;

        private static readonly string _appendTransportJSON = "{\"Transport\": \"",
            _appendTransportMessageJson = "\", \"Message\": ", _appendTransportEndJson = "}";


        private static IMessageBus GetMessageBus(WebSocketSubscriberInfo subscriberInfo) {
            var key = subscriberInfo.SubscriberName;
            var messageBus = _messageBuses.GetOrAdd(key,
                bus => GatewayMessageBus.Create(_esbEndpointAddress, new Core.Security.ESBIdentity(subscriberInfo.UserName, subscriberInfo.Password), false, subscriberInfo.Headers));
            if (messageBus != null && messageBus.IsClosed) {
                messageBus = _messageBuses[key] =
                    GatewayMessageBus.Create(_esbEndpointAddress, new Core.Security.ESBIdentity(subscriberInfo.UserName, subscriberInfo.Password), false, subscriberInfo.Headers);
            }
            return messageBus;    
        }

        internal static void CloseMessageBus() {
            foreach (var kv in _messageBuses) {
                var messageBus = kv.Value;
                if (messageBus != null && !messageBus.IsClosed)
                    messageBus.Dispose();
            }
            _messageBuses.Clear();
        }

        static internal void AddWebSocketConnection(IWebSocketConnection socket, WebSocketSubscriberInfo webSocketSubscriberInfo) {
            var webSocketKey = webSocketSubscriberInfo.SubscriberName;
            if (_connections.ContainsKey(socket.ID)) return;
            _connections[socket.ID] = webSocketKey;
            _webSockets[webSocketKey] = socket;

            if (!_handlers.ContainsKey(webSocketKey)) _handlers[webSocketKey] = new ConcurrentDictionary<string, dynamic>();

            if (!String.IsNullOrWhiteSpace(webSocketSubscriberInfo.ClientToken)) {
                _connectionTokens[webSocketKey] = webSocketSubscriberInfo.ClientToken;
                _connectionHeaders[webSocketKey] = webSocketSubscriberInfo.Headers;
            }
        }

        static internal void StreamMessages(WebSocketSubscriberInfo webSocketSubscriberInfo) {
            try {
                var subscriberName = webSocketSubscriberInfo.SubscriberName;
                var transportName = webSocketSubscriberInfo.TransportName;
                var webSocketKey = subscriberName;
                var socket = _webSockets[webSocketKey];
                if (socket == null) return;
                var messageBus = GetMessageBus(webSocketSubscriberInfo);
                var subscriber = messageBus.SelectSubscriber(subscriberName).ToSubscriber();
                var retryCount = webSocketSubscriberInfo.RetryCount;
                
                if (subscriber == null || !subscriber.IsValid || (subscriber.IsValid && !subscriber.Transports.Any(x => x.Key == transportName))) {
                    var reason = subscriber == null ? "Could not find subscriber" :
                        !subscriber.IsValid ? "Subscriber is in an invalid state" :
                        subscriber.IsValid && !subscriber.Transports.Any(x => x.Key == transportName) ?
                        String.Concat("Subscriber exists but could not find transport for ", transportName) :
                        string.Empty;
                    if (retryCount <= WEBSOCKET_RETRY_MAX)
                        socket.Send(new { NeedWebSocketRetry = true, Count = ++retryCount, Reason = reason, Transport = transportName }.ToJson());
                    else {
                        socket.Send(new { WebSocketRetryFailed = true, Reason = reason, Transport = transportName }.ToJson());
                    }
                    return;
                }
                var connected = true;
                dynamic handler = subscriber.OnMessageReceived<object>(transportName,
                    msg => 
                        {
                            var success = true;
                            WebStreamManagementManager.Execute(webMgmt => success = webMgmt.ShouldReceive(webSocketSubscriberInfo.ClientToken, webSocketSubscriberInfo.Headers));
                            if (!success) {
                                connected = false;
                                socket.Close();
                                return;
                            }
                            HandleMessage(transportName, socket, msg);
                        },
                    err => {
                        if (connected)
                            err.Continue = true;
                    },
                    TimeSpan.FromMilliseconds(Math.Max(webSocketSubscriberInfo.Interval, 1)),
                    webSocketSubscriberInfo.BatchSize);
                _handlers[webSocketKey][transportName] = handler;
            } catch (Exception ex) {
                ESBLogger.Log(ex);   
            }
        }

        private static void HandleMessage(string transport, IWebSocketConnection socket, object value) {
            try {
                socket.Send(String.Concat(_appendTransportJSON, transport, _appendTransportMessageJson, value.ToJson(), _appendTransportEndJson));
            } catch (Exception ex) { ESBLogger.Log(ex); }
        }
    }
}
